/*
 * @(#)BinlogTableThreadExec.java   1.0  2022年3月30日
 * 
 * Copyright (c)	2014-2020. All Rights Reserved.	GuangZhou hhmk Technology Company LTD.
 */
package com.swift.dts.source.thread;

import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

import javax.annotation.PostConstruct;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.alibaba.otter.canal.connector.core.consumer.CommonMessage;
import com.alibaba.otter.canal.connector.core.util.MessageUtil;
import com.alibaba.otter.canal.meta.exception.CanalMetaManagerException;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.ClientIdentity;
import com.alibaba.otter.canal.protocol.Message;
import com.swift.core.thread.SynThreadPoolExecutor;
import com.swift.dts.channel.ChannelManage;
import com.swift.dts.model.DtsBinlogMessage;
import com.swift.dts.source.canal.CanalDtsServer;
import com.swift.dts.util.CreateExecSql;
import com.swift.util.bean.BeanUtil;
import com.swift.util.exec.ThreadUtil;
import com.swift.util.type.JsonUtil;
import com.swift.util.type.TypeUtil;

/**
 * 添加说明
 * 
 * @author zhengjiajin
 * @version 1.0 2022年3月30日
 */
@Component
public class BinlogTableThreadExec {

    private static Logger log = LoggerFactory.getLogger(BinlogTableThreadExec.class);
    @Autowired
    private CanalDtsServer canalDtsServer;
    // getKey:thread(table)
    // private Map<String, TableExecThread> tableThread = new HashMap<>();
    // getKey:Queue(sql:10)
    private Map<String, LinkedBlockingQueue<ExecBinlog>> tableQueue = new ConcurrentHashMap<>();
    // getKey:Queue(Batch)
    // private Map<String, LinkedBlockingQueue<BatchAsk>> batchQueue = new ConcurrentHashMap<>();

    private PriorityBlockingQueue<BatchAsk> askBatchIdQueue;

    @Autowired
    private ChannelManage channelManage;

    private SynThreadPoolExecutor synThreadPoolExecutor;

    @PostConstruct
    private void init() {
        synThreadPoolExecutor = new SynThreadPoolExecutor(10, 200, 1);
        askBatchIdQueue = new PriorityBlockingQueue<BatchAsk>(11, new Comparator<BatchAsk>() {

            @Override
            public int compare(BatchAsk o1, BatchAsk o2) {
                if (o1.getBatchId() < o2.getBatchId()) {
                    return -1;
                }
                if (o1.getBatchId() > o2.getBatchId()) {
                    return 1;
                }
                return 0;
            }

        });

        Thread thread = new Thread(new Runnable() {

            public void run() {
                while (true) {
                    try {
                        BatchAsk ask = askBatchIdQueue.peek();
                        if (ask == null || ask.getSqlNum().get() > 0) {
                            ThreadUtil.sleep(1000);
                            continue;
                        }
                        ask = askBatchIdQueue.poll();
                        ack(ask.getClientIdentity(), ask.getBatchId());
                    } catch (Exception ex) {
                        log.error("ask binlog exception :", ex);
                    }
                }
            }
        });
        thread.start();
    }

    private void ack(ClientIdentity clientIdentity, long batchId) {
        try {
            canalDtsServer.ack(clientIdentity, batchId);
            log.info("ask binlog success :" + clientIdentity + ":" + batchId);
        } catch (CanalMetaManagerException ex) {
            log.error("ask binlog not first :"+ ex.getLocalizedMessage());
            //batchId:48 is not the firstly:37
            // 往前补提交
            String msg = ex.getLocalizedMessage();
            String firstId = msg.substring(msg.lastIndexOf(":")+1);
            if(TypeUtil.isNumber(firstId)) {
                long firstBatchId = TypeUtil.toLong(firstId);
                for(;firstBatchId<=batchId;firstBatchId++) {
                    canalDtsServer.ack(clientIdentity, firstBatchId);
                    log.info("ask binlog success :" + clientIdentity + ":" + batchId);
                }
            }
        }
    }
    
    /**
     * 如超出系统限制会出生阻塞
     * 
     * @param message
     */
    public void put(ClientIdentity clientIdentity, Message message) {
        if (message == null) return;
        if (message.getId() == -1) return;
        List<CanalEntry.Entry> entries = message.getEntries();
        entries.remove(null);
        List<CommonMessage> list = MessageUtil.convert(message);
        synchronized (clientIdentity.getDestination().intern()) {//需要暂时binlog
            BatchAsk batchAsk = new BatchAsk();
            batchAsk.setBatchId(message.getId());
            batchAsk.setClientIdentity(clientIdentity);
            batchAsk.setSqlNum(new AtomicInteger(list.size()));
            log.info("request binlog  :" + JsonUtil.toJson(batchAsk));
            askBatchIdQueue.add(batchAsk);
            for (CommonMessage common : list) {
                DtsBinlogMessage dtsMessage = new DtsBinlogMessage();
                BeanUtil.copyProperties(dtsMessage, common);
                dtsMessage.setDestination(clientIdentity.getDestination());
                log.info("收到BINLOG：" + JsonUtil.toJson(dtsMessage));
                String key = getPkKey(dtsMessage);
                synchronized (getLockKey(dtsMessage)) {
                    if (!tableQueue.containsKey(key)) tableQueue.put(key, new LinkedBlockingQueue<ExecBinlog>());
                    tableQueue.get(key).add(new ExecBinlog(batchAsk, dtsMessage));
                }
                synThreadPoolExecutor.execute(new TableExecThread(key));
            }
        }
    }

    private String getLockKey(CommonMessage message) {
        String lockStr = getPkKey(message)+"_LOCK";
        return lockStr.intern();
    }
    
    private String getPkKey(CommonMessage message) {
        return message.getDatabase() + "." + message.getTable() + "(" + CreateExecSql.getPkKey(message) + ")";
    }

    private class TableExecThread implements Runnable {

        private String key;

        private TableExecThread(String key) {
            this.key = key;
        }

        @Override
        public void run() {
            synchronized (key.intern()) {
                ExecBinlog binlog = tableQueue.get(key).poll();
                if (binlog == null) return;

                try {
                    channelManage.listener(binlog.getBtsBinlogMessage());
                } catch (Exception ex) {
                    log.error("channel exception :", ex);
                }
                binlog.getBatchAsk().getSqlNum().decrementAndGet();
                synchronized (getLockKey(binlog.getBtsBinlogMessage())) {
                    if (tableQueue.get(key).size() <= 0) tableQueue.remove(key);
                }
            }
        }

    }

    private static class ExecBinlog {

        private BatchAsk batchAsk;

        private DtsBinlogMessage btsBinlogMessage;

        private ExecBinlog(BatchAsk batchAsk, DtsBinlogMessage btsBinlogMessage) {
            this.batchAsk = batchAsk;
            this.btsBinlogMessage = btsBinlogMessage;
        }

        /**
         * @return the batchAsk
         */
        public BatchAsk getBatchAsk() {
            return batchAsk;
        }

        /**
         * @return the btsBinlogMessage
         */
        public DtsBinlogMessage getBtsBinlogMessage() {
            return btsBinlogMessage;
        }

    }

    private static class BatchAsk {

        private long batchId;

        private AtomicInteger sqlNum;

        private ClientIdentity clientIdentity;

        /**
         * @return the batchId
         */
        public long getBatchId() {
            return batchId;
        }

        /**
         * @param batchId
         *            the batchId to set
         */
        public void setBatchId(long batchId) {
            this.batchId = batchId;
        }

        /**
         * @return the sqlNum
         */
        public AtomicInteger getSqlNum() {
            return sqlNum;
        }

        /**
         * @param sqlNum
         *            the sqlNum to set
         */
        public void setSqlNum(AtomicInteger sqlNum) {
            this.sqlNum = sqlNum;
        }

        /**
         * @return the clientIdentity
         */
        public ClientIdentity getClientIdentity() {
            return clientIdentity;
        }

        /**
         * @param clientIdentity
         *            the clientIdentity to set
         */
        public void setClientIdentity(ClientIdentity clientIdentity) {
            this.clientIdentity = clientIdentity;
        }

    }
}
